CSA Data Uploaderで『ジョブ実行ジョブ』を使って1つ前のジョブ出力内容を入力ファイルとして扱う
はじめに
クラスメソッドが展開しているデータ統合基盤サービス『CSアナリティクス』(以降"CSA")のプロダクト群の1つ、『Data Uploader』には、 『予め作成してあるジョブ(=子ジョブ)を実行するためのジョブ(=親ジョブ)』である「ジョブ実行ジョブ」という機能があります。
この「ジョブ実行ジョブ」を使うと、1つ目のジョブの出力内容を2つ目のジョブの入力ファイルとして使うことができます。例えば
- 1つ目のジョブでローカルの当日日付(yyyy-mm-dd)のフォルダから読み込み、データの変換を行い、S3の当日日付(yyyy-mm-dd)のフォルダに出力
- 2つ目のジョブで、1つ目のジョブの出力結果を読み込み、データの変換を行い、S3の別の当日日付(yyyy-mm-dd)のフォルダに出力
のような処理です。図として書くと以下のようになります。
このような使い方をCSA Data Uploaderではどのように設定すればいいかについて紹介したいと思います。
なお今回「ジョブ実行ジョブ」「データアップロードジョブ」を主に使用しますが、詳細については以下のエントリも参照ください。
ジョブの作成
実際にジョブを作成していきたいと思います。子ジョブとなる「データアップロードジョブ」を2つ作成し、次に親ジョブとなる「ジョブ実行ジョブ」を作成していきたいと思います。
1つ目のデータアップロードジョブの作成
ジョブ一覧にて「ジョブを作成」を押下して1つ目のジョブを作成します。ジョブ名は「job_1」としました。
入力処理で読み込むファイルが存在することを確認します。今回はCSA Data Uploaderを実行するPCから読み込むので、想定するパスにファイルが存在することを確認します。
今回作成した「job_1」のフローは以下のようになります。ローカルからファイルを読み込む入力処理、データを変換する2つの変換処理、S3への出力処理となります。
入力処理を作成します。入力パスは以下のように「組み込み変数」（今回は「today」）を使い、当日日付のフォルダよりファイルを読み込むようにしました。変数として設定を行うと、パスイメージ上では設定した変数の形式で情報を置き換えて展開されていることが確認出来ます。
なおファイルパスが固定の場合は以下のように固定値でパスを指定することも可能です。
「データ表示」を押下するとデータのプレビューが表示されます。このファイルは以下のように「COL_A」〜「COL_K」の11列が存在します。
データの変換処理を追加します。今回はデータを変換・加工する例として、「列選択」にて末尾の列を出力項目から外してみました。
2つ目の変換処理を追加します。こちらも「列選択」にて末尾の列を出力項目から外してみました。
1つ目のジョブ「job_1」の出力するプレビューとなります。列が末尾から2つ削除されていることが分かると思います。
出力処理を作成します。出力パスは以下のように「組み込み変数」（今回は「today」）を使い、当日日付のフォルダにファイルを出力するようにしました。
出力処理についてもファイルパスが固定の場合は以下のように固定値でパスを指定することも可能です。
ジョブを実行します。正常に実行できると、指定したS3のパスにファイルが出力されます。
なお後述しますが、今回のように出力するファイルを次のジョブで使う場合は、ジョブを一度実行して出力結果のファイルを作成してください。
2つ目のデータアップロードジョブの作成
2つ目のジョブを作成していきます。ジョブ名は「job_2」としました。「job_2」のフローは以下のようになります。先に作成した「job_1」の出力を読み込む入力処理、2つの変換処理、S3への出力処理となります。
入力処理を作成します。入力ファイルは1つ目のジョブの出力ファイルとなりますが、ここでも下のように「組み込み変数」（今回は「today」）を使い、当日日付のフォルダよりファイルを読み込むようにしました。
なおファイルパスが固定の場合は以下のように固定値でパスを指定することも可能です。
「データ表示」を押下するとデータのプレビューが表示されます。入力ファイルはイメージは以下のようになります。
データの変換処理を追加します。先の「job_1」と同様、「列選択」にて末尾の列を出力項目から外してみました。
2つ目の変換処理を追加します。こちらも「列選択」にて末尾の列を出力項目から外してみました。
2つ目のジョブ「job_2」の出力するプレビューとなります。列が末尾から2つ削除されていることが分かると思います。
出力処理を作成します。出力パスは以下のように「組み込み変数」（今回は「today」）を使い、当日日付のフォルダにファイルを出力するようにしました。
出力処理についてもファイルパスが固定の場合は以下のように固定値でパスを指定することも可能です。
ジョブを実行してみます。正常に実行できると、指定したS3のパスにファイルが出力されます。
ジョブ実行ジョブの作成
ジョブ一覧にて「ジョブを作成」を押下して、子ジョブを2つ実行する親ジョブとなる「ジョブ実行ジョブ」を作成していきます。ジョブ名は「job_exec」としました。
先に作成した子ジョブの「job_1」「job_2」を追加します。
データアップロードジョブ、ファイルアップロードジョブの作成時に気をつけること
データアップロードジョブ・ファイルアップロードジョブ作成時に「保存」ボタン押下時、以下のチェックを行うようになりました。
- 入力するファイルが存在するか
- 変換処理が実際に実行可能かどうか
このためジョブを作成するときは予め入力ファイルを作成しておくことや、想定通りにジョブが実行できるかをプレビューで確認することを意識すると作業がやり易くなるかと思います。 上記のジョブ作成の例でも、「job_1」の作業前に入力ファイルが存在することを確認し、「job_2」の作業前には「job_1」を一度実行して入力ファイルが存在する状態で作業を行いました。
ジョブの実行
親ジョブとなる「job_exec」を実行してみます。
出力されたファイルをS3よりダウンロードし、解凍してファイルの中身を確認します。
子ジョブの「job_1」「job_2」の変換処理で指定したとおり、末尾から合計4列削除されたデータが出力されたことが分かるかと思います。
まとめ
ジョブ実行ジョブで1つ前のジョブ出力内容を入力ファイルとして扱う方法について見てみました。 これを使うと、例えば業務上の意味でジョブを複数に分け、それらを「ジョブ実行ジョブ」で順次実行させるといった使い方ができるかと思います。 何かの参考になれば幸いです。
